package org.apache.rocketmq.broker.client;

import io.netty.channel.Channel;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.util.PositiveAtomicCounter;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.body.ProducerInfo;
import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.store.stats.BrokerStatsManager;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;


/* @方法描述：生产者管理器
 */
public class ProducerManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    //客户端channel的过期时间
    private static final long CHANNEL_EXPIRED_TIMEOUT = 1000 * 120;

    //获取客户端channel的最大重试次数
    private static final int GET_AVAILABLE_CHANNEL_RETRY_COUNT = 3;

    private final ConcurrentHashMap<String /* group name */, ConcurrentHashMap<Channel, ClientChannelInfo>> groupChannelTable = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Channel> clientChannelTable = new ConcurrentHashMap<>();

    //Broker统计管理器，用于收集和报告各种统计数据
    protected final BrokerStatsManager brokerStatsManager;

    private PositiveAtomicCounter positiveAtomicCounter = new PositiveAtomicCounter();

    //存放生产者监听器的集合
    private final List<ProducerChangeListener> producerChangeListenerList = new CopyOnWriteArrayList<>();

    public ProducerManager() {
        this.brokerStatsManager = null;
    }

    public ProducerManager(final BrokerStatsManager brokerStatsManager) {
        this.brokerStatsManager = brokerStatsManager;
    }

    public int groupSize() {
        return this.groupChannelTable.size();
    }

    //判断指定主题的生产者客户端是否还在线
    public boolean groupOnline(String group) {
        //每次有客户端注销了，就会从groupChannelTable表中移除对应的客户端信息
        //如果这里得到的客户端信息不为null，说明还有对应的在线客户端
        Map<Channel, ClientChannelInfo> channels = this.groupChannelTable.get(group);
        return channels != null && !channels.isEmpty();
    }


    public ConcurrentHashMap<String, ConcurrentHashMap<Channel, ClientChannelInfo>> getGroupChannelTable() {
        return groupChannelTable;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到生产者客户端信息表的方法
     */
    public ProducerTableInfo getProducerTable() {
        Map<String, List<ProducerInfo>> map = new HashMap<>();

        //遍历groupChannelTable表，得到所有的主题信息，大家也可以理解为生产者组的组名
        for (String group : this.groupChannelTable.keySet()) {
            //得到每一个生产者组，这里得到的就是该主题下每一个生产者客户端的信息键值对
            for (Map.Entry<Channel, ClientChannelInfo> entry : this.groupChannelTable.get(group).entrySet()) {
                //得到当前遍历的生产者客户端的客户端信息
                ClientChannelInfo clientChannelInfo = entry.getValue();
                //判断当前遍历到的客户端的主题信息，也就是组名是否已经在map中存在了
                //如果存在那就可以直接把当前客户端信息添加到map对应的value中即可，不存在的话，还要为这个主题创建对应的list，然后把list添加到map中
                if (map.containsKey(group)) {
                    //如果map中已经有对应的list，那直接创建ProducerInfo对象，把该对象添加到list中即可
                    map.get(group).add(new ProducerInfo(
                            //设置当前客户端的唯一Id
                            clientChannelInfo.getClientId(),
                            //设置客户端的远程地址
                            clientChannelInfo.getChannel().remoteAddress().toString(),
                            //设置客户端的编程语言
                            clientChannelInfo.getLanguage(),
                            //设置客户端版本号
                            clientChannelInfo.getVersion(),
                            //设置客户端最新的更新时间
                            clientChannelInfo.getLastUpdateTimestamp()
                    ));
                } else {
                    //以下逻辑和上面分支中的类似，只不过多了一个创建list的操作，我就不添加注释了
                    map.put(group, new ArrayList<>(Collections.singleton(new ProducerInfo(
                            clientChannelInfo.getClientId(),
                            clientChannelInfo.getChannel().remoteAddress().toString(),
                            clientChannelInfo.getLanguage(),
                            clientChannelInfo.getVersion(),
                            clientChannelInfo.getLastUpdateTimestamp()
                    ))));
                }
            }
        }
        //返回ProducerTableInfo对象
        return new ProducerTableInfo(map);
    }

    /**
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：扫描并清理不活跃的客户端信息的方法
     */
    public void scanNotActiveChannel() {
        Iterator<Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>>> iterator = this.groupChannelTable.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry = iterator.next();
            //得到topic
            final String group = entry.getKey();
            //获取当前主题下所有客户端信息
            final ConcurrentHashMap<Channel, ClientChannelInfo> chlMap = entry.getValue();
            Iterator<Map.Entry<Channel, ClientChannelInfo>> it = chlMap.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry<Channel, ClientChannelInfo> item = it.next();
                //得到当前遍历的客户端对象
                final ClientChannelInfo info = item.getValue();
                //得到当前客户端最新活跃时间
                long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp();
                if (diff > CHANNEL_EXPIRED_TIMEOUT) {
                    it.remove();
                    //如果这个客户端已经不活跃了，则从groupChannelTable表中移除该客户端信息
                    Channel channelInClientTable = clientChannelTable.get(info.getClientId());
                    if (channelInClientTable != null && channelInClientTable.equals(info.getChannel())) {
                        clientChannelTable.remove(info.getClientId());
                    }
                    log.warn(
                            "ProducerManager#scanNotActiveChannel: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}",
                            RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group);
                    //执行监听器中的回调方法，触发的是客户端注销事件
                    callProducerChangeListener(ProducerGroupEvent.CLIENT_UNREGISTER, group, info);
                    //在这里把proxy服务端为客户端创建的channel关闭了
                    RemotingHelper.closeChannel(info.getChannel());
                }
            }
            //检查完之后，判断该主题下是否还有活跃的客户端信息
            if (chlMap.isEmpty()) {
                log.warn("SCAN: remove expired channel from ProducerManager groupChannelTable, all clear, group={}", group);
                //如果没有则从外层迭代器中把这个生产者组的数据都移除了
                iterator.remove();
                //调用监听中的方法，这里触发的是生产者组注销事件
                callProducerChangeListener(ProducerGroupEvent.GROUP_UNREGISTER, group, null);
            }


        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：注册生产者客户端信息的方法，方法中的第一个参数虽然名称是group，实际上是当前生产者客户端可以发送消息的主题
     */
    public synchronized void registerProducer(final String group, final ClientChannelInfo clientChannelInfo) {
        ClientChannelInfo clientChannelInfoFound = null;

        //根据主题信息从groupChannelTable表中获得对应的channel信息缓存表
        ConcurrentHashMap<Channel, ClientChannelInfo> channelTable = this.groupChannelTable.get(group);
        if (null == channelTable) {
            //不存在则创建新的放到groupChannelTable中
            channelTable = new ConcurrentHashMap<>();
            this.groupChannelTable.put(group, channelTable);
        }
        //根据proxy为客户端创建的channel得到对应的客户端channel信息，这里大家可能会有疑惑，一个生产者客户端不管可以向多少主题发送消息
        //只要它和proxy服务端建立了连接，那么proxy服务端为它创建的channel不就是唯一的吗？只有一个呀，为什么要用一个map来存储这个生产者客户端信息
        //并且还让proxy服务端为它创建的channel作为键？这个问题的答案其实非常简单，因为可能不止有一个生产者客户端和proxy服务端建立了连接
        //可能还有很多其他生产者客户端也向相同的主题发送消息，并且也连接了相同的客户端，这个时候，同一主题下的客户端信息不就都可以放到channelTable表中了吗？
        //而且各自的key都不相同，因为proxy会为每一个客户端创建channel对象，这个应该很好理解吧？
        clientChannelInfoFound = channelTable.get(clientChannelInfo.getChannel());
        //接下来的逻辑就很简单了，如果没有找到对应的客户端信息，则说明是一个新的生产者连接
        if (null == clientChannelInfoFound) {
            //那就把新的客户端通道信息添加到channelTable表中
            channelTable.put(clientChannelInfo.getChannel(), clientChannelInfo);
            //同时在clientChannelTable中添加一条记录，key为客户端ID，value为对应的客户端channel
            clientChannelTable.put(clientChannelInfo.getClientId(), clientChannelInfo.getChannel());
            log.info("new producer connected, group: {} channel: {}", group, clientChannelInfo.toString());
        }

        //如果找到了，说明这个注册生产者的操作是处理心跳请求时产生的
        if (clientChannelInfoFound != null) {
            //那就在这里更新客户端信息的最新更新时间，防止该客户端信息过期
            clientChannelInfoFound.setLastUpdateTimestamp(System.currentTimeMillis());
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：注销生产者客户端信息的方法
     */
    public synchronized void unregisterProducer(final String group, final ClientChannelInfo clientChannelInfo) {
        //根据主题信息得到对应的客户端信息缓存表
        ConcurrentHashMap<Channel, ClientChannelInfo> channelTable = this.groupChannelTable.get(group);
        //判断该主题下是否还有对应在线的客户端
        if (null != channelTable && !channelTable.isEmpty()) {
            //如果还有客户端，那就先把要注销的客户端的channel从表中移除
            ClientChannelInfo old = channelTable.remove(clientChannelInfo.getChannel());
            //根据要注销的客户端Id从clientChannelTable表中移除对应的数据
            clientChannelTable.remove(clientChannelInfo.getClientId());
            //判断是否移除成功
            if (old != null) {
                log.info("unregister a producer[{}] from groupChannelTable {}", group,
                        clientChannelInfo.toString());
                //执行生产者变化监听器中的方法，触发事件为客户端注销事件
                callProducerChangeListener(ProducerGroupEvent.CLIENT_UNREGISTER, group, clientChannelInfo);
            }

            //判断channelTable是否为空
            if (channelTable.isEmpty()) {
                //如果channelTable为空，就意味着该主题下没有活跃的客户端了
                //从groupChannelTable表中移除对应数据即可
                this.groupChannelTable.remove(group);
                //触发生产者组注销事件
                callProducerChangeListener(ProducerGroupEvent.GROUP_UNREGISTER, group, null);
                log.info("unregister a producer group[{}] from groupChannelTable", group);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到存活的生产者客户端channel的方法
     */
    public Channel getAvailableChannel(String groupId) {
        //对传入的参数判空，为空则返回null
        if (groupId == null) {
            return null;
        }
        List<Channel> channelList;
        //得到指定主题下所有的生产者客户端信息
        ConcurrentHashMap<Channel, ClientChannelInfo> channelClientChannelInfoHashMap = groupChannelTable.get(groupId);
        if (channelClientChannelInfoHashMap != null) {
            //如果数据不为空，那就返回所有的客户端channel
            channelList = new ArrayList<>(channelClientChannelInfoHashMap.keySet());
        } else {
            log.warn("Check transaction failed, channel table is empty. groupId={}", groupId);
            return null;
        }

        //接下来就可以根据具体的策略从channelList集合中选择一个channel返回出去了
        //这里得到集合的大小，显然接下来就要使用轮询的方式选择channel
        int size = channelList.size();
        if (0 == size) {
            log.warn("Channel list is empty. groupId={}", groupId);
            return null;
        }
        //要返回的备选channel对象
        Channel lastActiveChannel = null;

        //使用原子计数器来获取一个递增的值，并对channel集合大小取模，得到要选择的channel的索引
        int index = positiveAtomicCounter.incrementAndGet() % size;
        //根据索引从集合中的到对应的channel
        Channel channel = channelList.get(index);
        //失败重试次数
        int count = 0;
        //判断现在得到的这个channel是否活跃并且可写
        boolean isOk = channel.isActive() && channel.isWritable();
        //如果不OK那就在最大循环次数中得到可用的channel
        while (count++ < GET_AVAILABLE_CHANNEL_RETRY_COUNT) {
            //channel可用则直接返回
            if (isOk) {
                return channel;
            }
            //如果channel是活跃的，那就把这个channel记录下来，当作最后的备选方案
            if (channel.isActive()) {
                lastActiveChannel = channel;
            }
            //计算新的索引
            index = (++index) % size;
            //根据索引选择新的channel
            channel = channelList.get(index);
            //判断channel是否活跃和可写
            isOk = channel.isActive() && channel.isWritable();
        }
        //不管怎么说，最后返回的channel有两种情况，一种是null，也就是循环之后并没有找到活跃并且可写的channel
        //一种是活跃但并不可写的channel
        return lastActiveChannel;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端channel关闭事件的方法，该方法就不添加注释了，都是对成员变量map中数据的操作，重复性内容大家自己看看就行
     */
    public synchronized boolean doChannelCloseEvent(final String remoteAddr, final Channel channel) {
        boolean removed = false;
        if (channel != null) {
            for (final Map.Entry<String, ConcurrentHashMap<Channel, ClientChannelInfo>> entry : this.groupChannelTable.entrySet()) {
                final String group = entry.getKey();
                final ConcurrentHashMap<Channel, ClientChannelInfo> clientChannelInfoTable = entry.getValue();
                final ClientChannelInfo clientChannelInfo = clientChannelInfoTable.remove(channel);
                if (clientChannelInfo != null) {
                    clientChannelTable.remove(clientChannelInfo.getClientId());
                    removed = true;
                    log.info(
                            "NETTY EVENT: remove channel[{}][{}] from ProducerManager groupChannelTable, producer group: {}",
                            clientChannelInfo.toString(), remoteAddr, group);
                    callProducerChangeListener(ProducerGroupEvent.CLIENT_UNREGISTER, group, clientChannelInfo);
                    if (clientChannelInfoTable.isEmpty()) {
                        ConcurrentHashMap<Channel, ClientChannelInfo> oldGroupTable = this.groupChannelTable.remove(group);
                        if (oldGroupTable != null) {
                            log.info("unregister a producer group[{}] from groupChannelTable", group);
                            callProducerChangeListener(ProducerGroupEvent.GROUP_UNREGISTER, group, null);
                        }
                    }
                }
            }
        }
        return removed;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据客户端Id查找指定客户端channel的方法
     */
    public Channel findChannel(String clientId) {
        return clientChannelTable.get(clientId);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：执行所有生产者变化监听器的方法
     */
    private void callProducerChangeListener(ProducerGroupEvent event, String group,
                                            ClientChannelInfo clientChannelInfo) {
        for (ProducerChangeListener listener : producerChangeListenerList) {
            try {
                listener.handle(event, group, clientChannelInfo);
            } catch (Throwable t) {
                log.error("err when call producerChangeListener", t);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：注册生产者变化监听器的方法
     */
    public void appendProducerChangeListener(ProducerChangeListener producerChangeListener) {
        //其实就是把监听器添加到监听器列表中
        producerChangeListenerList.add(producerChangeListener);
    }
}
